package com.chukun.reactor.master;

import java.io.EOFException;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.nio.charset.StandardCharsets;

/**
 * @author chukun@meituan.com
 * @version 1.0.0
 * @description 异步处理器
 * @create 2022/1/25 21:23
 **/
public class AsyncHandler implements Runnable {

    private SocketChannel socketChannel;
    private SelectionKey sk;

    private ByteBuffer inputBuffer = ByteBuffer.allocate(1024);

    private ByteBuffer outputBuffer = ByteBuffer.allocate(1024);

    //存储客户端的完整消息
    private StringBuilder builder = new StringBuilder();

    public AsyncHandler(SocketChannel socketChannel) {
        this.socketChannel = socketChannel;
    }

    public SocketChannel getSocketChannel() {
        return socketChannel;
    }

    public void setSk(SelectionKey sk) {
        this.sk = sk;
    }

    @Override
    public void run() {
        try {
            if (sk.isReadable()) {
                read();
            } else if (sk.isWritable()) {
                write();
            }
        } catch (Exception e) {
            try {
                this.sk.channel().close();
            } catch (IOException ioException) {
                ioException.printStackTrace();
            }
        }
    }

    /**
     * 处理读请求
     *
     * @throws IOException
     */
    private void read() throws IOException {
        inputBuffer.clear();
        int n = socketChannel.read(inputBuffer);
        if (inputBufferComplete(n)) {
            System.out.println(Thread.currentThread().getName() + ":Server端收到客户端的请求消息：" + builder.toString());
            outputBuffer.put(builder.toString().getBytes(StandardCharsets.UTF_8));
            this.sk.interestOps(SelectionKey.OP_WRITE); //更改服务的逻辑状态以及处理的事件类型
        }
    }

    /**
     * 判断是否写入完成
     *
     * @param bytes
     * @return
     * @throws EOFException
     */
    private boolean inputBufferComplete(int bytes) throws EOFException {
        if (bytes > 0) {
            inputBuffer.flip(); //转化成读取模式
            while (inputBuffer.hasRemaining()) { //判断缓冲区中是否还有元素
                byte ch = inputBuffer.get(); //得到输入的字符
                if (ch == 3) { //表示Ctrl+c 关闭连接
                    throw new EOFException();
                } else if (ch == '\r' || ch == '\n') { //表示换行符
                    return true;
                } else {
                    builder.append((char) ch); //拼接读取到的数据
                }
            }
        } else if (bytes == -1) {
            throw new EOFException(); //客户端关闭了连接
        }
        return false;
    }

    /**
     * 处理写请求
     *
     * @throws IOException
     */
    private void write() throws IOException {
        int written = -1;
        //转化为读模式，判断是否有数据需要发送
        outputBuffer.flip();
        if (outputBuffer.hasRemaining()) {
            //把数据写回客户端
            written = socketChannel.write(outputBuffer);
        }
        outputBuffer.clear();
        builder.delete(0, builder.length());
        if (written <= 0) { //表示客户端没有输信息
            this.sk.channel().close();
        } else {
            socketChannel.write(ByteBuffer.wrap("\r\nreactor>".getBytes()));
            this.sk.interestOps(SelectionKey.OP_READ);
        }
    }

}
